package com.wjc.http.util;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.concurrent.Semaphore;

import android.util.Log;


/**
 * 
 * 处理服务器与客户端的通信
 * @author IO运行队列
 */
public class PipServer {
	private ArrayList<Element> list;
	private ArrayList<Element> list2;
	private ArrayList<Element> delete;
	private static final long TIME_OUT=10*1000; 
	private class Element{
		public Socket client;
		public Socket target;
		public long time1;
		public long time2;
		public Element(Socket client ,Socket target){
			this.client=client;
			this.target=target;
			time1=0;
			time2=0;
		}
	}
  private CheckRequestThread checkRequest;
  private CheckResponseThread checkResponse;
  private Semaphore sem1;//同步信号量
  private Semaphore sem2;
  private Semaphore SemL1;
  private Semaphore SemL2;
private Semaphore delSem;
  private class CheckRequestThread extends Thread{
	  private byte buffer[]=new byte[1024];
	  public void run(){
		  boolean flog;
		  while(true){
			  
			  if(list.size()==0){
				  sem1.acquireUninterruptibly();
				 
			  }
			  for(int i=0;i<list.size();i++){
				  SemL1.acquireUninterruptibly();
				  Element t=list.get(i);
				  SemL1.release();
				  flog=checkClient(t,buffer);
				  if(flog){
					  SemL1.acquireUninterruptibly();
					  list.remove(i);
					  SemL1.release();
					  delSem.acquireUninterruptibly();
					  if(delete.contains(t)){
						  delete.remove(t);
						  try {
							t.client.close();
							 t.target.close();
						} catch (IOException e) {
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
						 
					  }else{
						  delete.add(t);
					  }
					  delSem.release();
					  i--;
					  Log.d("停止请求", ""+(i+1));
					 
					  
				  }
				  
			  }
			  
			  
			  
		  }
		  
	  }
  } 
  private class CheckResponseThread extends Thread{
	  private byte buffer[]=new byte[1024];

		public void run() {
			boolean flog;
			while (true) {

				if (list2.size() == 0) {
					sem2.acquireUninterruptibly();

				}
				for (int i = 0; i < list2.size(); i++) {
					SemL2.acquireUninterruptibly();
					Element t = list2.get(i);
					SemL2.release();
					flog = checkTarget(t, buffer);
					if (flog) {
						SemL2.acquireUninterruptibly();
						list2.remove(i);
						SemL2.release();
						delSem.acquireUninterruptibly();
						if (delete.contains(t)) {
							delete.remove(t);
							try {
								t.client.close();
								t.target.close();
							} catch (IOException e) {
								// TODO Auto-generated catch block
								e.printStackTrace();
							}

						} else {
							delete.add(t);
						}
						delSem.release();
						i--;
						Log.d("停止响应", "" + (i + 1));

					}
					
				}

			}

		}
  } 
  public  PipServer(){
	  list=new ArrayList<Element>();
	  list2=new ArrayList<Element>();
	  delete=new ArrayList<Element>();
	  sem2=new Semaphore(0);
	  sem1=new Semaphore(0);
	  SemL1=new Semaphore(1);
	  SemL2=new Semaphore(1);
	  delSem=new Semaphore(1);
	  checkRequest=new CheckRequestThread();
	  checkRequest.setName("Pipe CheckRequest");
	  checkRequest.start();
	  checkResponse=new CheckResponseThread();
	  checkResponse.setName("Pipe CheckResponse");
	  checkResponse.start();
	  

  }
  
	public boolean checkTarget(Element t,byte []buf) {
		InputStream in;
		OutputStream out;
		try {
			if(t.target==null||t.client==null)return true;
			
			if(t.time2==0)t.time2=System.currentTimeMillis();
			out =t.client.getOutputStream();
			in = t.target.getInputStream();
			while(true){
			int size=in.available();
			
			if(size<=0){
				if(System.currentTimeMillis()-t.time2>TIME_OUT)
					return true;
				return false;
			}
			int length=in.read(buf);
			out.write(buf, 0, length);
			Log.d("response"+Thread.currentThread().getId(), "长度"+length);
			out.flush();
			t.time2=System.currentTimeMillis();
			}
			} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return true;
		}
		
}

	public boolean checkClient(Element t,byte[] buf) {
		InputStream in;
		OutputStream out;
		try {
			if(t.target==null||t.client==null)return true;
			if(t.time1==0)t.time1=System.currentTimeMillis();
			in = t.client.getInputStream();
			out =t.target.getOutputStream();
			while(true){
			int size=in.available();
			if(size<=0){
				if(System.currentTimeMillis()-t.time1>TIME_OUT)
					return true;
				return false;
			}
			int length=in.read(buf);
			out.write(buf, 0, length);
			Log.d("request"+Thread.currentThread().getId(), "长度"+length);
			out.flush();
			t.time1=System.currentTimeMillis();
			
			}
			} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return true;
		}
	
}

	public  void addSocket(Socket client,Socket target){
		Element t=new Element(client,target);
		if(client==null||target==null)return ;
		SemL1.acquireUninterruptibly();
		list.add(t);
		if(list.size()==1)sem1.release();
		SemL1.release();
		
		SemL2.acquireUninterruptibly();
		list2.add(t);
		if(list2.size()==1)sem2.release();
		SemL2.release();
		
		
	}
	
 
}  

